package com.partsoft.dits.utils;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.sql.DataSource;

import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.CallableStatementCallback;
import org.springframework.jdbc.core.ConnectionCallback;

import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.DataAction;
import com.partsoft.dits.message.DataProperty;
import com.partsoft.dits.utils.JdbcTemplate;

import org.springframework.jdbc.core.ResultSetExtractor;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

public class DbUtils {

	private static final ConcurrentHashMap<DataSource, PlatformTransactionManager> globalTransactionManagers = new ConcurrentHashMap<DataSource, PlatformTransactionManager>(5);

	/**
	 * 获得事务管理器，如果没有则创建
	 * 
	 * @param ds
	 *            数据源对象
	 * @return 对象管理器
	 */
	public static PlatformTransactionManager resolveTransactionManager(DataSource ds) {
		Assert.notNull(ds);
		PlatformTransactionManager txmanager = globalTransactionManagers.get(ds);
		if (globalTransactionManagers.containsKey(ds)) {
			txmanager = globalTransactionManagers.get(ds);
		} else {
			DataSourceTransactionManager dtxMgr = new DataSourceTransactionManager(ds);
			txmanager = globalTransactionManagers.putIfAbsent(ds, dtxMgr);
			if (txmanager == null) {
				txmanager = dtxMgr;
			}
		}
		return txmanager;
	}

	/**
	 * 移除事务管理器
	 * 
	 * @param ds
	 *            数据源对象
	 */
	public static void removeTransactionManager(DataSource ds) {
		Assert.notNull(ds);
		globalTransactionManagers.remove(ds);
	}

	/**
	 * 获得数据库连接所属的数据库产品名称
	 * 
	 * @param dsName
	 *            数据源
	 * @return 产品名
	 */
	public static String getDatabaseProduceName(Connection conn) {
		Assert.notNull(conn);
		try {
			if (conn.isClosed())
				throw new IllegalStateException("Connection is closed.");
			return conn.getMetaData().getDatabaseProductName() + "-" + conn.getMetaData().getDatabaseProductVersion();
		} catch (SQLException e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	/**
	 * 获得数据源所属的数据库产品名称
	 * 
	 * @param dsName
	 *            数据源
	 * @return 产品名
	 */
	public static String getDatabaseProduceName(DataSource ds) {
		Assert.notNull(ds);
		Connection conn = org.springframework.jdbc.datasource.DataSourceUtils.getConnection(ds);
		try {
			return getDatabaseProduceName(conn);
		} finally {
			org.springframework.jdbc.datasource.DataSourceUtils.releaseConnection(conn, ds);
		}
	}

	/**
	 * 获得数据源对应的所有的模式名称
	 * 
	 * @param ds
	 *            数据源
	 * @return 模式名称数组
	 */
	public static String[] getSchemaNames(final DataSource ds) {
		Assert.notNull(ds);
		return executeInReadonly(ds, new ReturnExecutionCallback<String[]>() {
			public String[] callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				List<String> schemaNames = jdbcHelper.execute(new ConnectionCallback<List<String>>() {
					public List<String> doInConnection(Connection connection) throws SQLException, DataAccessException {
						DatabaseMetaData dbmetadata = connection.getMetaData();
						ResultSet resultSet = dbmetadata.getSchemas();
						List<String> returnSchemas = new LinkedList<String>();
						while (resultSet.next()) {
							String schemaName = resultSet.getString(1);
							if (StringUtils.hasText(schemaName)) {
								returnSchemas.add(schemaName.trim());
							}
						}
						return returnSchemas;
					}
				});
				return schemaNames.toArray(new String[0]);
			}
		});
	}

	public static String packSchemaTableName(String schema, String tableName) {
		Assert.notNull(tableName);
		return StringUtils.hasText(schema) ? schema.trim() + "." + tableName : tableName;
	}

	public static boolean hasTableName(DataSource ds, String objectName) {
		return hasTableName(ds, null, objectName);
	}

	public static boolean hasTableName(final DataSource ds, final String schemaName, final String objectName) {
		Assert.notNull(ds);
		return executeInReadonly(ds, new ReturnExecutionCallback<Boolean>() {
			public Boolean callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				return jdbcHelper.execute(new ConnectionCallback<Boolean>() {
					public Boolean doInConnection(Connection connection) throws SQLException, DataAccessException {
						DatabaseMetaData dbmetadata = connection.getMetaData();
						ResultSet tableResult = dbmetadata.getTables(null, StringUtils.hasText(schemaName) ? schemaName
								: "%", objectName, new String[] { "TABLE", "VIEW" });
						return tableResult.next();
					}
				});
			}
		});
	}

	/**
	 * 通过数据源名称和模式名称返回所有表名
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @return 表名
	 */
	public static String[] getTableNames(final DataSource ds, final String schemaName) {
		Assert.notNull(ds);
		return executeInReadonly(ds, new ReturnExecutionCallback<String[]>() {
			public String[] callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				List<String> tableNames = jdbcHelper.execute(new ConnectionCallback<List<String>>() {
					public List<String> doInConnection(Connection connection) throws SQLException, DataAccessException {
						DatabaseMetaData dbmetadata = connection.getMetaData();
						ResultSet tableResult = dbmetadata.getTables(null, schemaName, "%", new String[] { "TABLE" });
						List<String> tNames = new LinkedList<String>();
						while (tableResult.next()) {
							String tbn = tableResult.getString("TABLE_NAME");
							if (StringUtils.hasText(tbn)) {
								tbn = tbn.trim();
								tNames.add(tbn);
							}
						}

						return tNames;
					}
				});

				return tableNames.toArray(new String[0]);
			}
		});
	}

	/**
	 * 通过数据源名称和模式名称返回所有视图名称
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @return 表名
	 */
	public static String[] getViewNames(final DataSource ds, final String schemaName) {
		Assert.notNull(ds);
		return executeInReadonly(ds, new ReturnExecutionCallback<String[]>() {
			public String[] callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				List<String> tableNames = jdbcHelper.execute(new ConnectionCallback<List<String>>() {
					public List<String> doInConnection(Connection connection) throws SQLException, DataAccessException {
						DatabaseMetaData dbmetadata = connection.getMetaData();
						ResultSet tableResult = dbmetadata.getTables(null, schemaName, "%", new String[] { "VIEW" });
						List<String> tNames = new LinkedList<String>();
						while (tableResult.next()) {
							String tbn = tableResult.getString("TABLE_NAME");
							if (StringUtils.hasText(tbn)) {
								tbn = tbn.trim();
								tNames.add(tbn);
							}
						}

						return tNames;
					}
				});

				return tableNames.toArray(new String[0]);
			}
		});
	}

	/**
	 * 通过数据源名称、模式名称、表名获得其所有的字段
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名称
	 * @return 字段数组
	 */
	public static DataProperty[] getTableColumns(final DataSource ds, final String schemaName, final String tableName) {
		Assert.notNull(ds);
		Assert.hasText(tableName, "未指定数据表名称");
		return executeInReadonly(ds, new ReturnExecutionCallback<DataProperty[]>() {
			public DataProperty[] callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				List<DataProperty> columnNames = jdbcHelper.execute(new ConnectionCallback<List<DataProperty>>() {
					public List<DataProperty> doInConnection(Connection connection) throws SQLException,
							DataAccessException {
						DatabaseMetaData dbmetadata = connection.getMetaData();
						ResultSet cResult = dbmetadata.getColumns(null, schemaName, tableName, "%");
						List<DataProperty> cNames = new LinkedList<DataProperty>();
						while (cResult.next()) {
							String cln = cResult.getString("COLUMN_NAME");
							int columnType = cResult.getInt("DATA_TYPE");
							int columnSize = cResult.getInt("COLUMN_SIZE");
							if (StringUtils.hasText(cln)) {
								cNames.add(new DataProperty(cln, columnType, columnSize));
							}
						}
						return cNames;
					}
				});
				if (columnNames.size() == 0) {
					columnNames = jdbcHelper.execute(
							String.format("select * from %s", packSchemaTableName(schemaName, tableName)),
							new CallableStatementCallback<List<DataProperty>>() {
								public List<DataProperty> doInCallableStatement(CallableStatement callablestatement)
										throws SQLException, DataAccessException {
									ResultSetMetaData res = callablestatement.getMetaData();
									List<DataProperty> cNames = new LinkedList<DataProperty>();
									for (int i = 1; i < res.getColumnCount() + 1; i++) {
										String cln = res.getColumnName(i);
										int columnType = res.getColumnType(i);
										int columnSize = res.getColumnDisplaySize(i);
										if (StringUtils.hasText(cln)) {
											cNames.add(new DataProperty(cln, columnType, columnSize));
										}
									}
									return cNames;
								}

							});
				}
				return columnNames.toArray(new DataProperty[0]);
			}
		});
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, Map<DataProperty, String> keyMap) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyMap, MessageDataItemResultFilter.NOP);
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, Map<DataProperty, String> keyMap, DataAction action) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyMap, new DataActionAppendResultFilter(action));
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, Map<DataProperty, String> keyMap, boolean notFoundPackKeys) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyMap, notFoundPackKeys,
				MessageDataItemResultFilter.NOP);
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, Map<DataProperty, String> keyMap, boolean notFoundPackKeys, DataAction action) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyMap, notFoundPackKeys,
				new DataActionAppendResultFilter(action));
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, Map<DataProperty, String> keyMap, MessageDataItemResultFilter filter) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyMap, false, filter);
	}

	public static int countRecordsByKey(DataSource ds, String schemaName, String tableName, DataProperty keyColumn,
			String value) {
		Assert.notNull(keyColumn);
		Assert.hasText(value);
		Map<DataProperty, String> keyMaps = new HashMap<DataProperty, String>(1);
		keyMaps.put(keyColumn, value);
		return countRecordsByKey(ds, schemaName, tableName, keyMaps);
	}

	public static int countRecordsByKey(final DataSource ds, String schemaName, String tableName,
			Map<DataProperty, String> keyMap) {
		Assert.notNull(ds);
		Assert.hasText(tableName);
		Assert.notEmpty(keyMap);

		final List<Object> paramValues = new ArrayList<Object>(keyMap.size());
		final StringBuffer queryStrBuffer = new StringBuffer("SELECT COUNT(*) FROM ");
		queryStrBuffer.append(packSchemaTableName(schemaName, tableName)).append(" WHERE ");
		boolean appended = false;
		for (Map.Entry<DataProperty, String> key : keyMap.entrySet()) {
			if (appended) {
				queryStrBuffer.append(" AND ");
			}
			queryStrBuffer.append(key.getKey().getPropertyName()).append(" = ?");
			paramValues.add(SQLTypeUtils.toSqlParameterValue(key.getKey().getValueType(), key.getValue()));
		}

		return executeInReadonly(ds, new ReturnExecutionCallback<Integer>() {
			public Integer callback() {
				JdbcTemplate jdbchelper = new JdbcTemplate(ds);
				return jdbchelper.queryForInt(queryStrBuffer.toString(), paramValues.toArray(new Object[0]));
			}
		});
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(final DataSource ds, final String schemaName,
			final String tableName, final Map<DataProperty, String> keyMap, boolean notFoundPackKeys,
			final MessageDataItemResultFilter filter) {
		Assert.notNull(ds);
		Assert.hasText(tableName);
		Assert.notEmpty(keyMap);
		final List<Object> paramValues = new ArrayList<Object>(keyMap.size());
		final StringBuffer queryStrBuffer = new StringBuffer("SELECT * FROM ");
		queryStrBuffer.append(packSchemaTableName(schemaName, tableName)).append(" WHERE ");
		boolean appended = false;
		for (Map.Entry<DataProperty, String> key : keyMap.entrySet()) {
			if (appended) {
				queryStrBuffer.append(" AND ");
			}
			queryStrBuffer.append(key.getKey().getPropertyName()).append(" = ?");
			paramValues.add(SQLTypeUtils.toSqlParameterValue(key.getKey().getValueType(), key.getValue()));
		}

		List<Map<DataProperty, String>> resultList = executeInReadonly(ds,
				new ReturnExecutionCallback<List<Map<DataProperty, String>>>() {
					public List<Map<DataProperty, String>> callback() {
						JdbcTemplate jdbchelper = new JdbcTemplate(ds);
						return new ArrayList<Map<DataProperty, String>>(jdbchelper.query(queryStrBuffer.toString(),
								paramValues.toArray(new Object[0]), new TableColumnMapMapper(keyMap.keySet(), filter)));
					}
				});
		if (resultList.size() == 0 && notFoundPackKeys) {
			DataProperty[] props = getTableColumns(ds, schemaName, tableName);
			Map<DataProperty, String> resultDatas = new HashMap<DataProperty, String>(props.length);
			for (DataProperty prop : props) {
				resultDatas.put(prop, "");
			}
			resultDatas.putAll(keyMap);
			if (filter != null)
				filter.filter(resultDatas);
			resultList.add(resultDatas);
		}
		return resultList;
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, DataProperty keyColumn, String sid) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyColumn, sid, false);
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, DataProperty keyColumn, String sid, DataAction action) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyColumn, sid, new DataActionAppendResultFilter(action));
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, DataProperty keyColumn, String sid, boolean notFoundPackKey) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyColumn, sid, notFoundPackKey,
				MessageDataItemResultFilter.NOP);
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, DataProperty keyColumn, String sid, MessageDataItemResultFilter filter) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyColumn, sid, false, filter);
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, DataProperty keyColumn, String sid, boolean notFoundPackKey,
			MessageDataItemResultFilter filter) {
		Assert.notNull(keyColumn);
		Map<DataProperty, String> keyMap = new HashMap<DataProperty, String>(1);
		keyMap.put(keyColumn, sid);
		return getTableRecordsByKey(ds, schemaName, tableName, keyMap, notFoundPackKey, filter);
	}

	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, String keyFieldName, int keyFieldType, String sid) {
		return getTableRecordsByKey(ds, schemaName, tableName, keyFieldName, keyFieldType, sid, false);
	}

	/**
	 * 根据主键返回符合条件的记录列表。
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名
	 * @param keyFieldName
	 *            主键字段名
	 * @param keyFieldType
	 *            主键类型({@link Types})
	 * @param sid
	 *            主键（字符，通过keyFieldType参数进行自动转换）
	 * @param notFoundPackKey
	 *            如果true在未找到记录的情况下把主键压入Map（其他字段为空）
	 * @return 记录列表
	 */
	public static List<Map<DataProperty, String>> getTableRecordsByKey(DataSource ds, String schemaName,
			String tableName, String keyFieldName, int keyFieldType, String sid, boolean notFoundPackKey) {
		Assert.hasText(keyFieldName);
		return getTableRecordsByKey(ds, schemaName, tableName, new DataProperty(keyFieldName, keyFieldType), sid,
				notFoundPackKey);
	}

	/**
	 * 根据传入的主键字段名获得所有的主键列表
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名
	 * @param keyFieldName
	 *            主键字段名
	 * @return 主键列表
	 */
	public static List<String> getTableRecordKeys(final DataSource ds, final String schemaName, final String tableName,
			final String keyFieldName) {
		Assert.notNull(ds);
		Assert.hasText(tableName, "必须传入表名");
		return executeInReadonly(ds, new ReturnExecutionCallback<List<String>>() {
			public List<String> callback() {
				JdbcTemplate jdbchelper = new JdbcTemplate(ds);
				StringBuffer queryStrBuffer = new StringBuffer("SELECT ");
				queryStrBuffer.append(keyFieldName).append(" FROM ").append(packSchemaTableName(schemaName, tableName));
				return new ArrayList<String>(jdbchelper.query(queryStrBuffer.toString(), new TableKeysRowMapper()));
			}
		});
	}

	public static List<Map<DataProperty, String>> getTableRecords(DataSource ds, String schemaName, String tableName) {
		return getTableRecords(ds, schemaName, tableName, MessageDataItemResultFilter.NOP);
	}

	public static List<Map<DataProperty, String>> getTableRecords(DataSource ds, String schemaName, String tableName,
			DataAction action) {
		return getTableRecords(ds, schemaName, tableName, new DataActionAppendResultFilter(action));
	}

	/**
	 * 获得表中所有记录
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名称
	 * @return 表记录Map
	 */
	public static List<Map<DataProperty, String>> getTableRecords(final DataSource ds, String schemaName,
			String tableName, final MessageDataItemResultFilter filter) {
		Assert.notNull(ds);
		Assert.hasText(tableName, "必须传入表名");
		final StringBuffer queryStrBuffer = new StringBuffer("SELECT * FROM ");
		queryStrBuffer.append(packSchemaTableName(schemaName, tableName));
		return executeInReadonly(ds, new ReturnExecutionCallback<List<Map<DataProperty, String>>>() {
			public List<Map<DataProperty, String>> callback() {
				JdbcTemplate jdbchelper = new JdbcTemplate(ds);
				return new ArrayList<Map<DataProperty, String>>(jdbchelper.query(queryStrBuffer.toString(),
						new TableColumnMapMapper(filter)));
			}
		});
	}

	/**
	 * 根据传入的主键值判断记录是否存在
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名称
	 * @param keyMap
	 *            主键Map
	 * @return true表示存在
	 */
	public static boolean isRecordKeyInTable(final DataSource ds, final String schemaName, final String tableName,
			final Map<DataProperty, String> keyMap) {
		Assert.notNull(ds, "必须传入数据源");
		Assert.hasText(tableName, "必须传入表名");
		Assert.notEmpty(keyMap, "主键不能为空");

		final StringBuffer zSQL = new StringBuffer("SELECT COUNT(*) FROM ");
		zSQL.append(packSchemaTableName(schemaName, tableName)).append(" WHERE ");
		final List<Object> paramValues = new ArrayList<Object>(keyMap.size());
		boolean appended = false;
		for (DataProperty cln : keyMap.keySet()) {
			if (appended) {
				zSQL.append(" AND ");
			}
			appended = true;
			zSQL.append(cln.getPropertyName()).append(" = ?");
			paramValues.add(SQLTypeUtils.toSqlParameterValue(cln.getValueType(), keyMap.get(cln)));
		}

		return executeInReadonly(ds, new ReturnExecutionCallback<Boolean>() {
			public Boolean callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				int keyCount = jdbcHelper.queryForInt(zSQL.toString(), paramValues.toArray(new Object[0]));
				if (keyCount >= 1) {
					if (keyCount > 1)
						Log.warn(String.format("数据表(%s)主键(%s)字段存在重复情况(%s)",
								packSchemaTableName(schemaName, tableName), keyMap.keySet().toString(), keyMap.values()
										.toString()));
					return true;
				}
				return false;
			}
		});
	}

	/**
	 * 根据传入的主键判断记录是否存在
	 * 
	 * @param ds
	 *            数据源句柄
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名称
	 * @param keyColumn
	 *            主键字段
	 * @param keyValue
	 *            主键数据
	 * @return true表示存在
	 */
	public static boolean isRecordKeyInTable(DataSource ds, String schemaName, String tableName, DataProperty column,
			String value) {

		Assert.notNull(ds, "必须传入数据源");
		Assert.notNull(column, "字段必须传入");
		Map<DataProperty, String> keyMap = new HashMap<DataProperty, String>(1);
		keyMap.put(column, value);
		return isRecordKeyInTable(ds, schemaName, tableName, keyMap);

	}

	/**
	 * 保存记录至数据库中
	 * 
	 * @param ds
	 * @param schemaName
	 * @param tableName
	 * @param defaultAction
	 * @param keyColumn
	 * @param keyValue
	 * @param row
	 */
	public static void saveRecordToTable(DataSource ds, String schemaName, String tableName, DataAction defaultAction,
			DataProperty keyColumn, String keyValue, Map<DataProperty, String> row) {

		Map<DataProperty, String> keyMap = new HashMap<DataProperty, String>(1);
		keyMap.put(keyColumn, keyValue);
		saveRecordToTable(ds, schemaName, tableName, defaultAction, keyMap, row);

	}

	/**
	 * 保存记录至数据库中
	 * 
	 * @param ds
	 *            数据源
	 * @param schemaName
	 *            模式名称
	 * @param tableName
	 *            表名称
	 * @param keyMap
	 *            主键Map
	 * @param row
	 *            一条记录字段
	 */
	public static void saveRecordToTable(final DataSource ds, String schemaName, String tableName,
			DataAction defaultAction, Map<DataProperty, String> keyMap, Map<DataProperty, String> row) {
		Assert.notNull(ds);
		Assert.hasText(tableName, "必须传入表名");

		if (defaultAction == null)
			defaultAction = DataAction.INSERT;
		Assert.isTrue(defaultAction != DataAction.GET, "SQL中不能使用DataAction.GET操作");
		if (defaultAction != DataAction.INSERT && CollectionUtils.isEmpty(keyMap)) {
			throw new IllegalArgumentException("主键参数[keyMap]为空时，不能执行为删除或更新动作");
		}

		boolean keyInTable = CollectionUtils.isEmpty(keyMap) ? false : isRecordKeyInTable(ds, schemaName, tableName,
				keyMap);
		if (defaultAction == DataAction.DELETE && keyInTable == false)
			return;
		if (defaultAction == DataAction.INSERT || defaultAction == DataAction.UPDATE)
			Assert.notEmpty(row, String.format("找不到要进行[%s]操作的字段", defaultAction.name()));

		if (keyInTable && defaultAction == DataAction.INSERT)
			defaultAction = DataAction.UPDATE;
		else if (keyInTable == false && defaultAction == DataAction.UPDATE)
			defaultAction = DataAction.INSERT;

		final List<Object> sqlParams = new LinkedList<Object>();
		final StringBuffer tSQL = new StringBuffer();
		StringBuffer wSQL = new StringBuffer();

		if (defaultAction == DataAction.INSERT) {
			tSQL.append("INSERT INTO ").append(packSchemaTableName(schemaName, tableName)).append("(");
			wSQL.append(" VALUES(");
			boolean appended = false;
			for (Map.Entry<DataProperty, String> fieldEntry : row.entrySet()) {
				if (ObjectUtils.nullSafeEquals(fieldEntry.getKey(), DataProperty.DATA_ACTION))
					continue;
				if (appended) {
					tSQL.append(",");
					wSQL.append(",");
				}
				appended = true;
				tSQL.append(fieldEntry.getKey().getPropertyName());
				wSQL.append("?");
				sqlParams.add(SQLTypeUtils.toSqlParameterValue(fieldEntry.getKey().getValueType(),
						fieldEntry.getValue()));
			}
			tSQL.append(")");
			wSQL.append(")");
		} else if (defaultAction == DataAction.UPDATE) {
			tSQL.append("UPDATE ").append(packSchemaTableName(schemaName, tableName)).append(" SET ");
			boolean appended = false;
			for (Map.Entry<DataProperty, String> fieldEntry : row.entrySet()) {
				if (keyMap.containsKey(fieldEntry.getKey()))
					continue;
				if (appended) {
					tSQL.append(", ");
				}
				appended = true;
				tSQL.append(fieldEntry.getKey().getPropertyName()).append(" = ? ");
				sqlParams.add(SQLTypeUtils.toSqlParameterValue(fieldEntry.getKey().getValueType(),
						fieldEntry.getValue()));
			}
			wSQL.append(" WHERE ");
			appended = false;
			for (Map.Entry<DataProperty, String> fieldEntry : keyMap.entrySet()) {
				if (appended) {
					wSQL.append(" AND ");
				}
				appended = true;
				wSQL.append(fieldEntry.getKey().getPropertyName()).append(" = ? ");
				sqlParams.add(SQLTypeUtils.toSqlParameterValue(fieldEntry.getKey().getValueType(),
						fieldEntry.getValue()));
			}
		} else if (defaultAction == DataAction.DELETE) {
			tSQL.append("DELETE FROM ");
			tSQL.append(packSchemaTableName(schemaName, tableName));
			wSQL.append(" WHERE ");
			boolean appended = false;
			for (Map.Entry<DataProperty, String> fieldEntry : keyMap.entrySet()) {
				if (appended) {
					wSQL.append(" AND ");
				}
				appended = true;
				wSQL.append(fieldEntry.getKey().getPropertyName()).append(" = ? ");
				sqlParams.add(SQLTypeUtils.toSqlParameterValue(fieldEntry.getKey().getValueType(),
						fieldEntry.getValue()));
			}
		}
		tSQL.append(wSQL.toString());

		executeInTransaction(ds, TransactionDefinition.PROPAGATION_REQUIRED, new ExecutionCallback() {
			public void callback() {
				JdbcTemplate jdbcHelper = new JdbcTemplate(ds);
				jdbcHelper.update(tSQL.toString(), sqlParams.toArray(new Object[0]));
			}
		});
	}

	/**
	 * 在使用DataSource的事务中执行回调（有返回值）
	 * 
	 * @param <T>
	 * @param ds
	 * @param trasAttr
	 *            {@link TransactionAttribute}
	 * @param callback
	 * @return
	 */
	public static <T> T executeInTransaction(DataSource ds, TransactionAttribute trasAttr,
			ReturnExecutionCallback<T> callback) {
		Assert.notNull(ds);
		Assert.notNull(trasAttr);
		PlatformTransactionManager txManager = resolveTransactionManager(ds);
		TransactionTemplate tt = new TransactionTemplate(txManager, trasAttr);
		final ReturnExecutionCallback<T> task = callback;
		return tt.execute(new TransactionCallback<T>() {
			public T doInTransaction(TransactionStatus transactionstatus) {
				return task.callback();
			}
		});
	}

	/**
	 * 在使用DataSource的事务中执行回调（有返回值）
	 * 
	 * @param <T>
	 * @param ds
	 * @param transactionPropagationBehavior
	 *            值内容：{@link TransactionDefinition}的静态属性
	 * @param callback
	 * @return
	 */
	public static <T> T executeInTransaction(DataSource ds, int transactionPropagationBehavior,
			ReturnExecutionCallback<T> callback) {
		Assert.notNull(ds);
		Assert.notNull(callback);
		return executeInTransaction(ds, new DefaultTransactionAttribute(transactionPropagationBehavior), callback);
	}

	/**
	 * 在只读事务中修改
	 * 
	 * @param <T>
	 * @param ds
	 * @param callback
	 * @return
	 */
	public static <T> T executeInReadonly(DataSource ds, ReturnExecutionCallback<T> callback) {
		Assert.notNull(ds);
		Assert.notNull(callback);
		DefaultTransactionAttribute tsAttr = new DefaultTransactionAttribute(TransactionDefinition.PROPAGATION_SUPPORTS);
		tsAttr.setReadOnly(true);
		return executeInTransaction(ds, tsAttr, callback);
	}

	/**
	 * 在使用DataSource的事务中执行回调（无返回值）
	 * 
	 * @param <T>
	 * @param ds
	 * @param trasAttr
	 *            {@link TransactionAttribute}
	 * @param callback
	 * @return
	 */
	public static void executeInTransaction(DataSource ds, TransactionAttribute trasAttr, ExecutionCallback callback) {
		Assert.notNull(ds);
		Assert.notNull(trasAttr);
		PlatformTransactionManager txManager = resolveTransactionManager(ds);
		TransactionTemplate tt = new TransactionTemplate(txManager, trasAttr);
		final ExecutionCallback task = callback;
		tt.execute(new TransactionCallback<Boolean>() {
			public Boolean doInTransaction(TransactionStatus transactionstatus) {
				task.callback();
				return Boolean.TRUE;
			}
		});
	}

	/**
	 * 在使用DataSource的事务中执行回调（无返回值）
	 * 
	 * @param <T>
	 * @param ds
	 * @param transactionPropagationBehavior
	 *            值内容：{@link TransactionDefinition}的静态属性
	 * @param callback
	 *            {@link ExecutionCallback}
	 * @return
	 */
	public static void executeInTransaction(DataSource ds, int transactionPropagationBehavior,
			ExecutionCallback callback) {
		Assert.notNull(ds);
		Assert.notNull(callback);
		executeInTransaction(ds, new DefaultTransactionAttribute(transactionPropagationBehavior), callback);
	}

	public static int countTableRecord(final DataSource ds, String schemaName, String tableName) {
		Assert.notNull(ds);
		Assert.hasText(tableName);
		final String tSQL = "SELECT COUNT(*) FROM " + packSchemaTableName(schemaName, tableName);
		return executeInReadonly(ds, new ReturnExecutionCallback<Integer>() {
			public Integer callback() {
				JdbcTemplate jdbctemplate = new JdbcTemplate(ds);
				return jdbctemplate.queryForInt(tSQL);
			}
		});
	}

	public static List<Map<DataProperty, String>> getTableRecords(final DataSource ds, String schemaName,
			String tableName, final int fromIndex, final int count) {
		return getTableRecords(ds, schemaName, tableName, fromIndex, count, MessageDataItemResultFilter.NOP);
	}

	public static List<Map<DataProperty, String>> getTableRecords(final DataSource ds, String schemaName,
			String tableName, final int fromIndex, final int count, DataAction action) {
		return getTableRecords(ds, schemaName, tableName, fromIndex, count, new DataActionAppendResultFilter(action));
	}

	@SuppressWarnings("unchecked")
	public static List<Map<DataProperty, String>> getTableRecords(final DataSource ds, String schemaName,
			String tableName, final int fromIndex, final int count, MessageDataItemResultFilter filter) {
		Assert.notNull(ds);
		Assert.hasText(tableName);
		Assert.isTrue(count > 0);
		Assert.isTrue(fromIndex >= 0);
		int recordCount = countTableRecord(ds, schemaName, tableName);
		if ((fromIndex + 1) > recordCount)
			return Collections.EMPTY_LIST;

		final String tSQL = "SELECT * FROM " + packSchemaTableName(schemaName, tableName);
		final TableColumnMapMapper columnMapper = new TableColumnMapMapper(filter);
		return executeInReadonly(ds, new ReturnExecutionCallback<List<Map<DataProperty, String>>>() {
			public List<Map<DataProperty, String>> callback() {
				JdbcTemplate jdbctemp = new JdbcTemplate(ds);
				return jdbctemp.query(tSQL, new ResultSetExtractor<List<Map<DataProperty, String>>>() {
					public List<Map<DataProperty, String>> extractData(ResultSet resultset) throws SQLException,
							DataAccessException {
						if (fromIndex > 0)
							try {
								resultset.absolute(fromIndex);
							} catch (SQLException e) {
								int i = 0;
								while (++i < fromIndex && resultset.next())
									;
							}
						int readc = 0;
						List<Map<DataProperty, String>> list = new LinkedList<Map<DataProperty, String>>();
						while (resultset.next() && readc < count) {
							list.add(columnMapper.mapRow(resultset, readc++));
						}
						return list;
					}
				});
			}
		});
	}

	public static class TableColumnMapMapper implements RowMapper<Map<DataProperty, String>> {

		private Map<String, DataProperty> columns = new HashMap<String, DataProperty>(5);

		private MessageDataItemResultFilter filter;

		public TableColumnMapMapper(Collection<DataProperty> columns, MessageDataItemResultFilter filter) {
			if (!CollectionUtils.isEmpty(columns)) {
				for (DataProperty cln : columns) {
					this.columns.put(cln.getPropertyName(), cln);
				}
			}
			this.filter = filter;
		}

		public TableColumnMapMapper(MessageDataItemResultFilter filter) {
			this(null, filter);
		}

		public TableColumnMapMapper(Collection<DataProperty> columns) {
			this(columns, null);
		}

		protected DataProperty getColumn(String columnName, Integer columnType, Integer columnSize) {
			columnName = columnName == null ? "" : columnName.trim().toUpperCase();
			if (columns.containsKey(columnName)) {
				DataProperty cln = columns.get(columnName);
				cln.setValueType(columnType);
				cln.setValueSize(columnSize);
			}
			DataProperty cln = new DataProperty(columnName, columnType, columnSize);
			columns.put(columnName, cln);
			return cln;
		}

		public Map<DataProperty, String> mapRow(ResultSet rs, int rowNum) throws SQLException {
			ResultSetMetaData metadata = rs.getMetaData();
			Map<DataProperty, String> objmap = new HashMap<DataProperty, String>(metadata.getColumnCount());
			for (int i = 1; i <= metadata.getColumnCount(); i++) {
				String columnName = metadata.getColumnName(i);
				String value = SQLTypeUtils.getXMLEncodingStringValue(rs, columnName, metadata.getColumnType(i));
				objmap.put(getColumn(columnName, metadata.getColumnType(i), metadata.getColumnDisplaySize(i)), value);
			}
			if (filter != null)
				filter.filter(objmap);
			return objmap;
		}
	}

	public static class TableKeysRowMapper implements RowMapper<String> {

		public String mapRow(ResultSet rs, int rowNum) throws SQLException {
			ResultSetMetaData metadata = rs.getMetaData();
			return SQLTypeUtils.getStringValue(rs, metadata.getColumnName(1), metadata.getColumnType(1));
		}
	}
	
	static class DataActionAppendResultFilter implements MessageDataItemResultFilter {
		
		private final DataAction action;
		
		public DataActionAppendResultFilter(DataAction action) {
			this.action = action;
			Assert.notNull(action);
		}
		
		public void filter(Map<DataProperty, String> dataResult) {
			if (!CollectionUtils.isEmpty(dataResult)) {
				dataResult.put(DataProperty.DATA_ACTION, action.name());
			}
		}
		
	}
}
